package com.markhsiu.MiniMQ.remote.transport.netty;

import java.util.concurrent.TimeUnit;

import com.markhsiu.MiniMQ.client.CallBackCache;
import com.markhsiu.MiniMQ.client.CallBackor;
import com.markhsiu.MiniMQ.client.Client;
import com.markhsiu.MiniMQ.model.Message;
import com.markhsiu.MiniMQ.model.Result;
import com.markhsiu.MiniMQ.remote.Address;
import com.markhsiu.MiniMQ.remote.transport.netty.handler.NettyClientHandler;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

/**
 * Created by Mark Hsiu on 2017/2/8.
 */
public class NettyClient implements Client {

	private EventLoopGroup worker = new NioEventLoopGroup();
	private Bootstrap bootstrap;

	private Address addr;
	private Channel messageChannel;
	private CallBackor callBackor;

	public NettyClient() {
		addr = new Address();
	}

	public NettyClient(Address serverURL) {
		if (serverURL == null) {
			throw new RuntimeException(" server address is null");
		}
		addr = serverURL;
	}

	@Override
	public Result send(Message message) {
		final Result result = new Result();

		CallBackCache.putCallBack(message.getMessageID(), result);
		ChannelFuture channel = messageChannel.writeAndFlush(message);
		channel.addListener(new ChannelFutureListener() {
			public void operationComplete(ChannelFuture future) throws Exception {
				if (!future.isSuccess()) {
					System.out.println(future.cause());
					result.setError(1);
				}
			}
		});

		Message ask = result.getMessageResult(10, TimeUnit.SECONDS);
		if (ask == null) {
			if (result.getError() == 1) {
				throw new RuntimeException("响应超时");
			} else {
				System.out.println("结果为null");
			}
		}

		result.setMessage(ask);
		return result;
	}

	
	@Override
	public void setCallBackor(CallBackor callBackor) {
		this.callBackor = callBackor;
	}

	@Override
	public Address getServerAddr() {
		return addr;
	}

	@Override
	public void open() {
		bootstrap = new Bootstrap();
		bootstrap.group(worker).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true)
				.option(ChannelOption.SO_KEEPALIVE, false).handler(new ChilderChannelHandler());

	}

	@Override
	public void connect() {
		if (bootstrap == null) {
			throw new RuntimeException(" you need open Client before connect");
		}

		try {
			ChannelFuture future = bootstrap.connect(addr.getHostname(), addr.getPort()).sync();
			future.addListener(new ChannelFutureListener() {
				public void operationComplete(ChannelFuture future) throws Exception {
					Channel channel = future.channel();
					messageChannel = channel;
				}
			});

		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void close() {
		try {
			if (messageChannel != null) {
				messageChannel.close().sync();
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			worker.shutdownGracefully();
		}

	}

	private class ChilderChannelHandler extends ChannelInitializer<SocketChannel> {

		@Override
		protected void initChannel(SocketChannel channel) throws Exception {
			ChannelPipeline pip = channel.pipeline();
			// 1. 禁止对类加载器进行缓存，它在基于OSGI的动态模块化编程中经常使用
			pip.addLast(new ObjectDecoder(1024, ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
			pip.addLast(new ObjectEncoder());

			pip.addLast(new NettyClientHandler(callBackor));
		}
	}

}
